package scalaapps

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.graphx.GraphLoader

/**
  * Created by yang on 16-12-30.
  */
object PR2 {

  def main(args: Array[String]) {

    val conf = new SparkConf()
    conf.setAppName("PageRank").setMaster("local[2]")
    val sc = new SparkContext(conf)
    // 从特定的边列表文件中读取数据生成图框架
    val graph = GraphLoader.edgeListFile(sc, "/home/yang/data/follows")
    // 用上面的图框架来调用pageRank(动态)算法
    //特别注意：静态调用的方法名是staticPageRank(Int)
    // vertices将返回顶点属性
    val ranks = graph.pageRank(0.00001).vertices
    // 将上面得到的ranks（顶点属性）和用户进行关系连接
    // 首先也是读取一个包含了用户信息的文件，然后调用了一个map函数，即将文件里的每行数据按 ”,” 切开并返回存储了处理后数据的RDD
    val users = sc.textFile("/home/yang/data/users").map { line =>
      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }
    // 这里具体实现了将ranks和用户列表一一对应起来
    // 从map函数的内容可以看出是按id来进行连接，但返回的结果只含用户名和它的相应rank值
    val ranksByUsername = users.join(ranks).map {
      case (id, (username, rank)) => (username, rank)
    }
    // 收集上面RDD里的数据并打印出来
    println(ranksByUsername.collect().mkString("\n"))
  }

}
